package avicit.bdp.lineage.druid;


import avicit.bdp.lineage.druid.analyse.SqlRequestContext;
import avicit.bdp.lineage.druid.analyse.SqlResponseContext;
import avicit.bdp.lineage.druid.analyse.handler.DefaultHandlerChain;
import avicit.bdp.lineage.druid.analyse.handler.FillTableExpressHandler;
import avicit.bdp.lineage.druid.analyse.handler.LineageTableHandler;
import avicit.bdp.lineage.druid.exception.ParserException;
import avicit.bdp.lineage.druid.model.ColumnNode;
import avicit.bdp.lineage.druid.model.FieldLineageModel;
import avicit.bdp.lineage.druid.model.FieldNameModel;
import avicit.bdp.lineage.druid.model.TableLineageModel;
import avicit.bdp.lineage.druid.model.TableNameModel;
import avicit.bdp.lineage.druid.model.TableNode;
import avicit.bdp.lineage.druid.model.TreeNode;
import avicit.bdp.lineage.druid.util.LineageContext;
import com.alibaba.druid.DbType;
import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.ast.statement.SQLDropTableStatement;
import com.alibaba.druid.sql.ast.statement.SQLExprTableSource;
import com.alibaba.druid.sql.visitor.SchemaStatVisitor;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * <p>
 * SQL 解析
 * </p>
 */

public class LineageSqlUtils {
    private static final Logger logger = LoggerFactory.getLogger(LineageSqlUtils.class);

    private static LineageSqlUtils instance;
    private DefaultHandlerChain defaultHandlerChain;

    private LineageSqlUtils() {
        defaultHandlerChain = new DefaultHandlerChain();
        defaultHandlerChain.addHandler(new LineageTableHandler());
        defaultHandlerChain.addHandler(new FillTableExpressHandler());
    }

    public static LineageSqlUtils getInstance() {
        if (instance == null) {
            instance = new LineageSqlUtils();
        }
        return instance;
    }

    /**
     * SQL 解析
     *
     * @param sql
     * @param dbType 数据库类型
     * @return avicit.bdp.lineage.druid.util.LineageContext
     */
    public LineageContext handle(String sql, String dbType) {
        LineageContext context = new LineageContext();
        sql = sql.trim().toLowerCase();
        SqlRequestContext requestContext = new SqlRequestContext();
        requestContext.setSql(sql);
        requestContext.setDbType(dbType);
        this.handleSql(requestContext, context);

        return context;
    }

    /**
     * 多个 SQL 解析，支持里面有临时表，解析时会把临时表去掉
     *
     * @param sqlList
     * @param dbType  数据库类型
     * @return avicit.bdp.lineage.druid.util.LineageContext
     */
    public LineageContext handle(List<String> sqlList, String dbType) {
        LineageContext result = new LineageContext();
        //解析临时表
        Map<String, String> tmpTableMap = handleTmpTables(sqlList, dbType);

        //处理SQL解析,处理字段血缘
        List<FieldLineageModel> fieldNodeList = new ArrayList<>();
        for (String sql : sqlList) {
            sql = sql.trim().toLowerCase();
            if (StringUtils.isBlank(sql)) {
                continue;
            }
            try {
                LineageContext context = new LineageContext();
                SqlRequestContext requestContext = new SqlRequestContext();
                requestContext.setSql(sql);
                requestContext.setDbType(dbType);
                this.handleSql(requestContext, context);
                fieldNodeList.addAll(context.getFieldNodeList());
            } catch (Exception e) {
                logger.error("执行错误SQL：", sql);
                logger.error(e.getMessage(), e);
            }

        }
        //字段血缘列表合并
        List<FieldLineageModel> mergedFieldNodeList = handleMergeFieldLineageModels(fieldNodeList);
        //去除关系中的临时表
        List<FieldLineageModel> removedFieldNodeList = handleRemoveTmpTable(mergedFieldNodeList, tmpTableMap);
        //设置字段血缘
        result.setFieldNodeList(removedFieldNodeList);
        //处理表关系
        handleTableRelation(result);

        return result;
    }

    /**
     * 移除字段血缘列表中的临时表
     *
     * @param fieldNodeList
     * @param tmpTableMap
     * @return java.util.List<avicit.bdp.lineage.druid.model.FieldLineageModel>
     */
    private List<FieldLineageModel> handleRemoveTmpTable(List<FieldLineageModel> fieldNodeList, Map<String, String> tmpTableMap) {
        if (tmpTableMap.isEmpty()) {
            return fieldNodeList;
        }
        //处理目标字段的临时表
        Map<String, FieldLineageModel> removeTableMap = new HashMap<>();
        List<FieldLineageModel> removedList = new ArrayList<>();
        List<FieldLineageModel> fieldLineageModelList = new ArrayList<>();
        for (FieldLineageModel model : fieldNodeList) {
            FieldNameModel fieldNameModel = model.getTargetField();
            if (tmpTableMap.containsKey(fieldNameModel.getTableName())) {
                removeTableMap.put(fieldNameModel.getTableName() + "-" + fieldNameModel.getFieldName(), model);
                removedList.add(model);
            } else {
                fieldLineageModelList.add(model);
            }
        }
        //递归处理临时表中来源字段的临时表
        for (FieldLineageModel model : removedList) {
            HashSet<FieldNameModel> subFields = handleRecursionTmpTable(model.getSourceFields(), removeTableMap);
            model.setSourceFields(subFields);
            FieldNameModel fieldNameModel = model.getTargetField();
            //把处理好的字段重新放入Map中
            removeTableMap.put(fieldNameModel.getTableName() + "-" + fieldNameModel.getFieldName(), model);
        }

        //处理来源字段临时表问题
        for (FieldLineageModel model : fieldLineageModelList) {
            HashSet<FieldNameModel> sourceFields = model.getSourceFields();
            HashSet<FieldNameModel> fields = new HashSet<>();
            for (FieldNameModel sourceField : sourceFields) {
                String key = sourceField.getTableName() + "-" + sourceField.getFieldName();
                if (removeTableMap.containsKey(key)) {
                    FieldLineageModel subModel = removeTableMap.get(key);
                    fields.addAll(subModel.getSourceFields());
                } else {
                    fields.add(sourceField);
                }
            }
            model.setSourceFields(fields);
        }

        return fieldLineageModelList;
    }

    private HashSet<FieldNameModel> handleRecursionTmpTable(HashSet<FieldNameModel> sourceFields, Map<String, FieldLineageModel> removeTableMap) {
        HashSet<FieldNameModel> fields = new HashSet<>();
        for (FieldNameModel sourceField : sourceFields) {
            String key = sourceField.getTableName() + "-" + sourceField.getFieldName();
            if (removeTableMap.containsKey(key)) {
                FieldLineageModel subModel = removeTableMap.get(key);
                //如果自己引用自己，则不处理
                if (!sourceField.equals(subModel.getTargetField())) {
                    HashSet<FieldNameModel> subFields = handleRecursionTmpTable(subModel.getSourceFields(), removeTableMap);
                    fields.addAll(subFields);
                }

            } else {
                fields.add(sourceField);
            }
        }
        return fields;
    }

    /**
     * 合并多个sql的字段血缘
     *
     * @param fieldNodeList
     * @return java.util.List<avicit.bdp.lineage.druid.model.FieldLineageModel>
     */
    private List<FieldLineageModel> handleMergeFieldLineageModels(List<FieldLineageModel> fieldNodeList) {
        List<FieldLineageModel> mergedFieldNodeList = new ArrayList<>();
        Map<String, FieldLineageModel> modelMap = new HashMap<>();
        for (FieldLineageModel model : fieldNodeList) {
            String key = model.getTargetField().getTableName() + "-" + model.getTargetField().getFieldName();
            if (!modelMap.containsKey(key)) {
                modelMap.put(key, model);
                mergedFieldNodeList.add(model);
            } else {
                FieldLineageModel fieldLineageModel = modelMap.get(key);
                fieldLineageModel.getSourceFields().addAll(model.getSourceFields());
            }
        }
        return mergedFieldNodeList;
    }

    /**
     * 解析临时表
     *
     * @param sqlList
     * @param dbType
     * @return java.util.Map<java.lang.String, java.lang.String>
     */
    private Map<String, String> handleTmpTables(List<String> sqlList, String dbType) {
        Map<String, String> tableMap = new HashMap<>();
        for (String sql : sqlList) {
            try {
                List<SQLStatement> stmtList = SQLUtils.parseStatements(sql, dbType);
                // 解析出的独立语句的个数
                for (int i = 0; i < stmtList.size(); i++) {

                    SQLStatement stmt = stmtList.get(i);
                    SchemaStatVisitor visitor = SQLUtils.createSchemaStatVisitor(DbType.of(dbType));
                    stmt.accept(visitor);
                    if (stmt instanceof SQLDropTableStatement) {
                        List<SQLExprTableSource> tableSourceList = ((SQLDropTableStatement) stmt).getTableSources();
                        for (SQLExprTableSource tableSource : tableSourceList) {
                            String tableName = getTableNameFromName(tableSource.toString());
                            tableMap.put(tableName.toLowerCase(), tableName.toLowerCase());
                        }

                    }
                }
            } catch (Exception e) {
                logger.error("SQL解析出错：" + e.getMessage());
            }

        }
        return tableMap;
    }

    /**
     * 获取表名
     *
     * @param name
     * @return java.lang.String
     */
    private String getTableNameFromName(String name) {
        if (name == null) {
            return null;
        }
        // 处理表名带引号"";
        name = name.replaceAll("\"", "");
        // 先处理是否有字段
        if (name.contains(".")) {
            String str[] = name.split("\\.", 2);
            name = str[1];
        } else if (name.contains("@")) {
            //处理oracle的实例名
            String str[] = name.split("@", 2);
            name = str[0];
        }

        return name;
    }

    /**
     * 处理SQL解析
     *
     * @param request
     * @param context
     * @return void
     */
    private void handleSql(SqlRequestContext request, LineageContext context) {
        FieldLineageModel currentFieldModel = null;

        SqlResponseContext response = new SqlResponseContext();
        defaultHandlerChain.handle(request, response);
        if (Objects.isNull(response.getLineageTableTree()) || Objects.isNull(response.getLineageTableTree().getValue())) {
            return;
        }
        //分析列血缘
        handleColumnRelation(request, response, context);

        //分析表血缘
        handleTableRelation(context);

    }

    /**
     * 处理字段血缘关系
     *
     * @param sqlContext
     * @param response
     * @param context
     * @return void
     */
    private void handleColumnRelation(SqlRequestContext sqlContext, SqlResponseContext response, LineageContext context) {
        TreeNode<TableNode> lineageTableTree = response.getLineageTableTree();
        TableNode rootTable = lineageTableTree.getValue();
        //目标表字段
        List<ColumnNode> rootColumns = rootTable.getColumns();
        //如果不存在，则从第一个有字段的表中获取
        if (CollectionUtils.isEmpty(rootColumns)) {
            TreeNode<TableNode> firstHaveColumnTableNode = this.findFirstHaveColumnTableNode(lineageTableTree);
            List<ColumnNode> firstColumns = firstHaveColumnTableNode.getValue().getColumns();
            //把目标表的表名设置成字段所在的表
            firstColumns.stream().forEach(node -> {
                ColumnNode columnNode = new ColumnNode();
                //针对部分 只有别名的列 ，select 1 as a1,2 as a2
                if (StringUtils.isBlank(node.getName())) {
                    columnNode.setName(node.getAlias());
                } else {
                    columnNode.setName(node.getName());
                }
                columnNode.setAlias(node.getAlias());
                columnNode.setTableName(rootTable.getName());
                columnNode.setOwner(rootTable);
                rootColumns.add(columnNode);
            });
        }

        if (CollectionUtils.isEmpty(rootColumns)) {
            throw new ParserException("列没有解析出来");
        }

        List<TreeNode<TableNode>> withList = lineageTableTree.getChildList();
        if (withList.size() > 1) {
            //存在with语法
            for (int i = 1; i < withList.size(); i++) {
                TreeNode<TableNode> withTableNode = withList.get(i);
                context.getWithTreeNodeMap().put(withTableNode.getValue().getAlias(), withTableNode);
            }

        }
        initPhysicalTableNodeMap(lineageTableTree, context);


        for (ColumnNode node : rootColumns) {
            //System.out.println("target: " + node.getTableSchemaName() + "  " + node.getTableName() + " " + node.getName());
            //每一列都初始化一次当前字段模型
            context.setCurrentFieldModel(new FieldLineageModel());

            FieldNameModel targetField = new FieldNameModel();
            targetField.setTableName(node.getTableName());
            //第一层，如果有别名，则以别名为主
            if (StringUtils.isNotBlank(node.getAlias())) {
                targetField.setFieldName(node.getAlias());
            } else {
                targetField.setFieldName(node.getName());
            }
            targetField.setDbName(node.getTableSchemaName());
            context.getCurrentFieldModel().setTargetField(targetField);
            traceColumnLineage(sqlContext.getDbType(), node, lineageTableTree.getChildList().get(0), 0, context);
            context.getFieldNodeList().add(context.getCurrentFieldModel());
            //System.out.println("########## ");
        }


    }

    /**
     * 处理表关系
     *
     * @param context
     * @return void
     */
    private void handleTableRelation(LineageContext context) {
        List<FieldLineageModel> fieldLineageModelList = context.getFieldNodeList();
        Map<String, TableLineageModel> targetCacheMap = new HashMap<>();
        Map<String, Object> sourceCacheMap = new HashMap<>();
        List<TableLineageModel> tableLineageModelList = new ArrayList<>();
        for (FieldLineageModel model : fieldLineageModelList) {
            FieldNameModel targetField = model.getTargetField();
            TableLineageModel targetLineageModel = targetCacheMap.get(targetField.getTableName());
            if (targetLineageModel == null) {
                targetLineageModel = new TableLineageModel();
                TableNameModel targetTable = new TableNameModel();
                targetTable.setTableName(targetField.getTableName());
                targetTable.setDbName(targetField.getDbName());
                targetLineageModel.setTargetTable(targetTable);
                targetCacheMap.put(targetField.getTableName(), targetLineageModel);
                tableLineageModelList.add(targetLineageModel);
            }
            HashSet<FieldNameModel> sourceFields = model.getSourceFields();
            HashSet<TableNameModel> sourceTables = targetLineageModel.getSourceTables();
            for (FieldNameModel sourceFiled : sourceFields) {
                if (!sourceCacheMap.containsKey(sourceFiled.getTableName())) {
                    TableNameModel tableNameModel = new TableNameModel();
                    tableNameModel.setTableName(sourceFiled.getTableName());
                    tableNameModel.setDbName(sourceFiled.getDbName());
                    sourceTables.add(tableNameModel);
                }
            }
        }
        context.setTableNodeList(tableLineageModelList);
    }

    /**
     * 初始化物理表名
     *
     * @param lineageTableTree
     * @return void
     */
    private void initPhysicalTableNodeMap(TreeNode<TableNode> lineageTableTree, LineageContext context) {
        List<TreeNode<TableNode>> childList = lineageTableTree.getChildList();
        for (TreeNode<TableNode> tableNode : childList) {
            TableNode subTableNode = tableNode.getValue();
            if (subTableNode.getVirtualTemp() == false && !(context.getWithTreeNodeMap().containsKey(subTableNode.getAlias())
                    || context.getWithTreeNodeMap().containsKey(subTableNode.getName()))) {
                if (StringUtils.isNotBlank(subTableNode.getName())) {
                    context.getTableNodeMap().put(subTableNode.getName(), subTableNode);
                }
                if (StringUtils.isNotBlank(subTableNode.getAlias())) {
                    context.getTableNodeMap().put(subTableNode.getAlias(), subTableNode);
                }
            }
            if (!tableNode.isLeaf()) {
                initPhysicalTableNodeMap(tableNode, context);
            }
        }
    }

    /**
     * 字段血缘解析
     *
     * @param dbType
     * @param currentColumn
     * @param tableNode
     * @param level
     * @return void
     */
    private void traceColumnLineage(String dbType, ColumnNode currentColumn,
                                    TreeNode<TableNode> tableNode, int level, LineageContext context) {
        //列的名称和别名
        String columnName = currentColumn.getName();
        String columnAlias = currentColumn.getAlias();
        String tableName = currentColumn.getTableName();
        String tableNameAlias = currentColumn.getTableNameAlas();
        if (StringUtils.isBlank(tableNameAlias)) {
            tableNameAlias = "";
        }

        TableNode subTableNode = tableNode.getValue();
        String subTableNameAlias = subTableNode.getAlias();
        List<ColumnNode> subColumnNodeList = subTableNode.getColumns();
        //如果当前的的表名和子查询的别名不一致，则不需要进一步的递归判断
        if (StringUtils.isNotBlank(subTableNameAlias) && (StringUtils.isNotBlank(tableName) && !tableName.equalsIgnoreCase(subTableNameAlias))) {
            return;
        }

        //列如果不为空，则进行列名比对
        if (!subColumnNodeList.isEmpty()) {
            //第一层，则判断从select中获取的列名一致就行, 表示insert对应的列名，从第二层开始，需要判断表名和列名是否一致
            //level是计算获取column的等级，不是节点等级
            for (ColumnNode subColumnNode : subColumnNodeList) {
                if (level == 0) {

                    compareColumn(currentColumn, dbType, subColumnNode, tableNode, level, context);

                } else {
                    //从第二层开始, 表示对当前虚拟表的字段，继续递归追踪，需要增加表名的比对

                    //找到列名所在表名  select t1.c1 from table t1
                    String columnOwnerTableName = subColumnNode.getTableName();
                    //如果没有对字段加别名，则补上表名    select c1 from table t1
                    //对于多列的表名 select (a.c1+b.c2) c3 处理, 此处不进行表名的获取
                    if (StringUtils.isBlank(columnOwnerTableName) &&
                            (subColumnNode.getSourceColumns() == null || subColumnNode.getSourceColumns().isEmpty())) {
                        columnOwnerTableName = getSubChildNodeTableAlias(tableNode);
                        if (StringUtils.isBlank(columnOwnerTableName)) {
                            columnOwnerTableName = this.repairMissingTableName(subColumnNode, dbType);
                        }

                        int schemaIndex = columnOwnerTableName.indexOf(".");
                        if (schemaIndex > 0) {
                            subColumnNode.setTableName(columnOwnerTableName.substring(schemaIndex + 1));
                            subColumnNode.setTableSchemaName(columnOwnerTableName.substring(0, schemaIndex));
                        } else {
                            subColumnNode.setTableName(columnOwnerTableName);
                        }

                    }
                    if (StringUtils.isNotBlank(tableNameAlias)) {
                        subColumnNode.setTableNameAlas(tableNameAlias);
                    }

                    //表名是否一致
                    if (isTableNameSame(tableName, tableNameAlias, columnOwnerTableName, subTableNameAlias)) {
                        compareColumn(currentColumn, dbType, subColumnNode, tableNode, level, context);
                    } else {
                        //继续递归判断
                        List<TreeNode<TableNode>> subTableNodeList = tableNode.getChildList();
                        if (subTableNodeList == null || subTableNodeList.isEmpty()) {
                            continue;
                        }
                        for (TreeNode<TableNode> subTableNodeTree : subTableNodeList) {
                            traceColumnLineage(dbType, currentColumn, subTableNodeTree, level + 1, context);
                        }

                    }

                }
            }
        } else {
            //如果是空的，则递归判断下级节点
            List<TreeNode<TableNode>> subTableNodeList = tableNode.getChildList();
            if (subTableNodeList == null || subTableNodeList.isEmpty()) {
                return;
            }
            for (TreeNode<TableNode> subTableNodeTree : subTableNodeList) {
                traceColumnLineage(dbType, currentColumn, subTableNodeTree, level, context);
            }

        }

    }

    /**
     * 表名和表别名比较
     *
     * @param tableName
     * @param tableNameAlias
     * @param subTableName
     * @param subTableNameAlias
     * @return boolean
     */
    private boolean isTableNameSame(String tableName, String tableNameAlias, String subTableName, String subTableNameAlias) {
        if (StringUtils.isNotBlank(tableName)) {
            if (tableName.equalsIgnoreCase(subTableName) || tableName.equalsIgnoreCase(subTableNameAlias)) {
                return true;
            }
        }
        if (StringUtils.isNotBlank(tableNameAlias)) {
            if (tableNameAlias.equalsIgnoreCase(subTableName) || tableNameAlias.equalsIgnoreCase(subTableNameAlias)) {
                return true;
            }
        }
        return false;
    }

    /**
     * 获取下级节点的别名，只取第一个 aa
     * select a1,b1 from (select a1,b1 from table1 V ) aa
     *
     * @param tableNode
     * @return java.lang.String
     */
    private String getSubChildNodeTableAlias(TreeNode<TableNode> tableNode) {
        List<TreeNode<TableNode>> subTableNodeList = tableNode.getChildList();
        if (subTableNodeList != null && !subTableNodeList.isEmpty()) {
            return subTableNodeList.get(0).getValue().getAlias();
        }
        return "";
    }

    /**
     * 比较来源列名，并递归比较
     *
     * @param currentColumn
     * @param dbType
     * @param subColumnNode
     * @param tableNode
     * @param level
     * @return void
     */
    private void compareColumn(ColumnNode currentColumn, String dbType, ColumnNode subColumnNode,
                               TreeNode<TableNode> tableNode, int level, LineageContext context) {

        //支持字段由多字段构成 select (c1+c2) as c3, 如果存在多字段，则以多字段为准
        List<ColumnNode> list = new ArrayList<>();
        if (currentColumn.getSourceColumns() != null && !currentColumn.getSourceColumns().isEmpty()) {
            list.addAll(currentColumn.getSourceColumns());
        } else {
            list.add(currentColumn);
        }
        //处理比较目标字段是多字段
        for (ColumnNode columnNode : list) {
            //列的名称和别名
            String columnName = columnNode.getName() == null ? "" : columnNode.getName();
            boolean flag;
            if(columnNode.getAlias() == null){
                flag = columnName.equalsIgnoreCase(subColumnNode.getName())
                        || columnName.equalsIgnoreCase(subColumnNode.getAlias())
                        || (subColumnNode.getName() != null && subColumnNode.getName().equalsIgnoreCase("*"));
            }else{
                flag = columnNode.getAlias().equalsIgnoreCase(subColumnNode.getAlias())
                        && (columnName.equalsIgnoreCase(subColumnNode.getName())
                        || columnName.equalsIgnoreCase(subColumnNode.getAlias())
                        || (subColumnNode.getName() != null && subColumnNode.getName().equalsIgnoreCase("*")));
            }

            //字段相同, 如果查询表为*，则默认表示该字段存在并且和比较字段一样的名称
            if (flag) {

                //如果没有对字段加别名，则补上表名    select c1 from table t1
                //先处理来源字段是多字段的问题类似 select (a.c1+a.c2) as c3 from t2
                List<ColumnNode> sourcelist = new ArrayList<>();
                if (subColumnNode.getSourceColumns() != null && !subColumnNode.getSourceColumns().isEmpty()) {
                    sourcelist.addAll(subColumnNode.getSourceColumns());
                } else {
                    sourcelist.add(subColumnNode);
                }

                //循环目标字段判断
                for (ColumnNode sourceColumnNode : sourcelist) {
                    //找到列名所在表名  select t1.c1 from table t1
                    String columnOwnerTableName = sourceColumnNode.getTableName();
                    if (StringUtils.isBlank(columnOwnerTableName)) {
                        //从下级节点获取别名
                        columnOwnerTableName = getSubChildNodeTableAlias(tableNode);
                        if (StringUtils.isBlank(columnOwnerTableName)) {
                            //从sql中获取表名
                            columnOwnerTableName = this.repairMissingTableName(subColumnNode, dbType);
                        }
                        int schemaIndex = columnOwnerTableName.indexOf(".");
                        if (schemaIndex > 0) {
                            sourceColumnNode.setTableName(columnOwnerTableName.substring(schemaIndex + 1));
                            sourceColumnNode.setTableSchemaName(columnOwnerTableName.substring(0, schemaIndex));
                            columnOwnerTableName = sourceColumnNode.getTableName();
                        } else {
                            sourceColumnNode.setTableName(columnOwnerTableName);
                        }
                    }
                    //判断表名是否是物理表
                    if (isPhysicalTable(sourceColumnNode.getTableName(), context)) {
                        TableNode node = context.getTableNodeMap().get(sourceColumnNode.getTableName());
                        //表示 select (c1+c2) as c from table t1

                        //是物理表，表示是当前字段的来源字段
                        FieldNameModel sourceField = new FieldNameModel();
                        sourceField.setTableName(node.getName());
                        //针对select * 的处理
                        if (StringUtils.isNotBlank(sourceColumnNode.getName()) && sourceColumnNode.getName().equalsIgnoreCase("*")) {
                            sourceField.setFieldName(columnName);
                        } else {
                            sourceField.setFieldName(sourceColumnNode.getName());
                        }
                        sourceField.setDbName(node.getSchemaName());
                        //如果字段名称为空，则可能是常量字段
                        if (StringUtils.isBlank(sourceField.getFieldName())) {
                            return;
                        }
                        //判断是否和目标字段是同一个字段，如果否，则加入
                        if (!sourceField.equals(context.getCurrentFieldModel().getTargetField())) {
                            context.getCurrentFieldModel().getSourceFields().add(sourceField);
                        }

                        //System.out.println(sourceField.getDbName() + "  " + sourceField.getTableName() + " " + sourceField.getFieldName());

                    } else if (context.getWithTreeNodeMap().containsKey(sourceColumnNode.getTableName())) {
                        TreeNode<TableNode> withTreeNode = context.getWithTreeNodeMap().get(sourceColumnNode.getTableName());
                        traceColumnLineage(dbType, sourceColumnNode, withTreeNode, 0, context);
                    } else {
                        List<TreeNode<TableNode>> subTableNodeList = tableNode.getChildList();
                        if (subTableNodeList == null || subTableNodeList.isEmpty()) {
                            return;
                        }
                        for (TreeNode<TableNode> subTableNodeTree : subTableNodeList) {
                            traceColumnLineage(dbType, sourceColumnNode, subTableNodeTree, level + 1, context);
                        }

                    }
                }


            } else {
                //不同， 不处理

            }
        }

    }

    private boolean isPhysicalTable(String tableName, LineageContext context) {
        return context.getTableNodeMap().containsKey(tableName);
    }


    /**
     * 找到第一个有字段的节点
     *
     * @param root TableNode
     * @return TreeNode<TableNode>
     */
    private TreeNode<TableNode> findFirstHaveColumnTableNode(TreeNode<TableNode> root) {
        if (!CollectionUtils.isEmpty(root.getValue().getColumns())) {
            return root;
        }
        if (CollectionUtils.isEmpty(root.getChildList())) {
            throw new ParserException("node.found.more");
        }
        TreeNode<TableNode> subTreeNode = root.getChildList().get(0);
        //TODO 待改进
        if (CollectionUtils.isEmpty(subTreeNode.getValue().getColumns())) {
            return findFirstHaveColumnTableNode(subTreeNode);
        } else {
            // 第一个有字段的节点，其父级仅有一个子元素
            return root.getChildList().get(0);
        }

    }

    /**
     * 修复缺失的表名
     * for example:
     * 1. select distinct a1,b1 from table1 V
     * 2. select distinct a1,b1 from table1,table2 X
     * 第一种情况识别可以，第二种会取第一个表的字段，建议在SQL书写时，指定出对应别名
     *
     * @param columnNode columnNode
     * @param dbType     dbType
     * @return 表名
     */
    private String repairMissingTableName(ColumnNode columnNode, String dbType) {
        if (StringUtils.isEmpty(columnNode.getTableExpression())) {
            throw new ParserException("repair missing table, table expression can't null.");
        }
        SQLStatement stmt = SQLUtils.parseSingleStatement(columnNode.getTableExpression(), dbType);
        SchemaStatVisitor schemaStatVisitor = new SchemaStatVisitor();
        stmt.accept(schemaStatVisitor);
        return schemaStatVisitor.getTables().keySet().stream().findFirst()
                .orElseThrow(() -> new ParserException("repair missing table failed,column expression[%s].",
                        columnNode.getExpression()))
                .getName();
    }


}
